package cn.xuexiyuan.flinkstudy.example;

import cn.xuexiyuan.flinkstudy.entity.OrderAggregateResult;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;

/**
 * @Description:
 * @Author 左龙龙
 * @Date 21-3-31
 * @Version 1.0
 **/
public class WindowResult implements WindowFunction<Integer, OrderAggregateResult, String, TimeWindow> {

    @Override
    public void apply(String keyword, TimeWindow window, Iterable<Integer> input, Collector<OrderAggregateResult> out) throws Exception {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");

        int total = input.iterator().next();
        OrderAggregateResult result = new OrderAggregateResult();
        result.setChannel_keyword(keyword);
        result.setDate(sdf.format(window.getStart()));
        result.setTotal(total);
        out.collect(result);

//        System.out.println(String.format("窗口时间： [%s - %s)", sdf.format(window.getStart()), sdf.format(window.getEnd())));
    }
}
